package com.hang.ad.mysql.listener;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.hang.ad.mysql.TemplateHolder;
import com.hang.ad.mysql.dto.BinlogRowData;
import com.hang.ad.mysql.dto.TableTemplate;
import com.sun.org.apache.bcel.internal.generic.IF_ACMPEQ;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.*;
import java.util.stream.Collectors;

@Component
@Slf4j
public class AggregationListener implements BinaryLogClient.EventListener {

    private String dbName;
    private String tableName;

    private Map<String, Ilistener> listenerMap = new HashMap<>();
    @Autowired
    private TemplateHolder templateHolder;

    public void register(String dbName, String tableName, Ilistener ilistener){
        this.listenerMap.put(dbName + ":" + tableName, ilistener);
    }
    @Override
    public void onEvent(Event event) {
        EventType type = event.getHeader().getEventType();
        if (type == EventType.TABLE_MAP){
            TableMapEventData data = event.getData();
            this.dbName = data.getDatabase();
            this.tableName = data.getTable();
            return;
        }

        if (type != EventType.UPDATE_ROWS && type != EventType.WRITE_ROWS && type != EventType.DELETE_ROWS){
            return;
        }

        if (StringUtils.isEmpty(dbName) || StringUtils.isEmpty(tableName)){
            log.error("no meta data");
            return;
        }

        Ilistener ilistener = this.listenerMap.get(this.dbName + ":" + this.tableName);
        if (ilistener == null){
            return;
        }

        log.info("trigger event :{}" , type.name());
        try {
            BinlogRowData binlogRowData = buildRowData(event.getData());
            if (binlogRowData == null){
                return;
            }
            binlogRowData.setEventType(type);
            ilistener.onEvent(binlogRowData);
        }catch (Exception e){
            log.error(e.getMessage());
        }finally {
            this.dbName = "";
            this.tableName = "";
        }
    }

    private List<Serializable[]> getAfterValues(EventData eventData){
        if (eventData instanceof WriteRowsEventData){
            return ((WriteRowsEventData) eventData).getRows();
        }
        if (eventData instanceof UpdateRowsEventData){
            return ((UpdateRowsEventData) eventData).getRows().stream().map(Map.Entry::getValue).collect(Collectors.toList());
        }
        if (eventData instanceof DeleteRowsEventData){
            return ((DeleteRowsEventData) eventData).getRows();
        }
        return Collections.emptyList();
    }
    private BinlogRowData buildRowData(EventData eventData){
        TableTemplate table = templateHolder.getTable(tableName);
        if (table == null){
            log.warn("table{} not found", tableName);
            return null;
        }

        List<Map<String,String>> afterList = new ArrayList<>();
        for (Serializable[] after : getAfterValues(eventData)) {
            Map<String,String> afterMap = new HashMap<>();
            int length = after.length;
            for (int i=0; i < length; i++){
                String colName = table.getPosMap().get(i);
                if (colName == null){
                    continue;
                }
                String colValue = after[i].toString();
                afterMap.put(colName, colValue);
            }
            afterList.add(afterMap);
        }

        BinlogRowData binlogRowData = new BinlogRowData();
        binlogRowData.setAfter(afterList);
        binlogRowData.setTable(table);

        return binlogRowData;
    }
}
